package big.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


public class HbaseApi {
    private static ThreadLocal<Connection> connHolder = new ThreadLocal<Connection>();

    public static Configuration conf = HBaseConfiguration.create();

    /**
     * 得到hbase连接对象
     * @return
     */
    public static Connection getConnection() throws IOException {
        Connection connection = connHolder.get();
        if (connection == null){
            connection = ConnectionFactory.createConnection(conf);
            connHolder.set(connection);
        }
        return connection;
    }

    /**
     * 关闭连接对象
     * @throws IOException
     */
    public static void close() throws IOException {
        Connection connection = connHolder.get();
        if (connection != null){
            connection.close();
            connHolder.remove();
        }
    }

    //判断表是否存在
    public static boolean isExitTable(String tableName) throws IOException {
        //从连接池中创建连接对象
        Connection connection =  connHolder.get();
        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
        //此对象对表进行操作
        boolean exists = admin.tableExists(tableName);
        return exists;
    }

    //创建表
    public static void createTable (String tableName, String... colFamly) throws IOException{
        //connHolder.set(ConnectionFactory.createConnection(conf));
        //Connection connection = connHolder.get();
        Connection connection = connHolder.get();
        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
        if(isExitTable(tableName)){
            System.out.println("表" + tableName + "已存在");
            //System.exit(0);
        }else{
            //创建表属性对象,表名需要转字节
            HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));
            //创建多个列族
            for(String cf : colFamly){
                descriptor.addFamily(new HColumnDescriptor(cf));
            }
            //根据对表的配置，创建表
            admin.createTable(descriptor);
            System.out.println("表" + tableName + "创建成功！");
        }
    }

    /**
     * 删除表
     * @param tableName 表名
     * @throws IOException
     */
    public static void deleteTable (String tableName) throws IOException{
        Connection connection =  connHolder.get();
        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
        //表管理对象
        if (!isExitTable(tableName)){
            System.out.println(tableName+"表不存在");
        }else{
            //先禁用表
            admin.disableTable(tableName);
            //删除表
            admin.deleteTable(tableName);
            System.out.println("删除表"+tableName+"成功");
        }
    }

    /**
     * 添加表数据
     * @param tableName
     * @param rowKey
     * @param famlyCol 列族
     * @param col 列明
     * @param value 值
     * @throws IOException
     */
    public static void addData(String tableName,String rowKey,String famlyCol,String col,String value) throws IOException {
        Connection connection = connHolder.get();
        if (!isExitTable(tableName)){
            System.out.println("该表不存在!");
        }else{
            Table hTable = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(famlyCol),Bytes.toBytes(col),Bytes.toBytes(value));
            hTable.put(put);
            System.out.println("插入成功!");
        }
    }

    /**
     * 删除多行表数据
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public static void deleteData(String tableName,String... rowKey) throws IOException {
        Connection connection = connHolder.get();
        if (!isExitTable(tableName)){
            System.out.println("该表不存在!");
        }else{
            Table hTable = connection.getTable(TableName.valueOf(tableName));
            List<Delete> deletes= new ArrayList<Delete>();
            for (String row :
                    rowKey) {
                Delete delete = new Delete(Bytes.toBytes(row));
                deletes.add(delete);
            }
            hTable.delete(deletes);
        }
    }

    /**
     * 扫描全表
     * @param tableName
     * @throws IOException
     */
    public static void getAllRows(String tableName) throws IOException {
        Connection connection = connHolder.get();
        if (!isExitTable(tableName)) {
            System.out.println("该表不存在!");
        } else {
            Table hTable = connection.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            ResultScanner results = hTable.getScanner(scan);
            for (Result rs :
                    results) {
                Cell[] cells = rs.rawCells();
                for (Cell cell : cells) {
                    //得到rowkey
                    System.out.println("行键:" + Bytes.toString(CellUtil.cloneRow(cell)));
                    //得到列族
                    System.out.println("列族" + Bytes.toString(CellUtil.cloneFamily(cell)));
                    System.out.println("列:" + Bytes.toString(CellUtil.cloneQualifier(cell)));
                    System.out.println("值:" + Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
        }
    }

    /**
     * 根据rowKey获取数据
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public static void getRowsByRowKey(String tableName,String rowKey) throws IOException{
            Connection connection = connHolder.get();
            if (!isExitTable(tableName)) {
                System.out.println("该表不存在!");
            } else {
                Table hTable = connection.getTable(TableName.valueOf(tableName));
                Get get = new Get(Bytes.toBytes(rowKey));
                Scan scan = new Scan(get);
                ResultScanner results = hTable.getScanner(scan);
                for (Result rs :
                        results) {
                    Cell[] cells = rs.rawCells();
                    for (Cell cell : cells) {
                        //得到rowkey
                        System.out.println("行键:" + Bytes.toString(CellUtil.cloneRow(cell)));
                        //得到列族
                        System.out.println("列族" + Bytes.toString(CellUtil.cloneFamily(cell)));
                        System.out.println("列:" + Bytes.toString(CellUtil.cloneQualifier(cell)));
                        System.out.println("值:" + Bytes.toString(CellUtil.cloneValue(cell)));
                    }
                }
            }
    }

    /**
     * 根据rowkey,列族,列明获取数据
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @throws IOException
     */
    public static void getRowQualifier(String tableName, String rowKey, String family, String
            qualifier) throws IOException{
        Connection connection = connHolder.get();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        Result result = table.get(get);
        for(Cell cell : result.rawCells()){
            System.out.println("行键:" + Bytes.toString(result.getRow()));
            System.out.println("列族" + Bytes.toString(CellUtil.cloneFamily(cell)));
            System.out.println("列:" + Bytes.toString(CellUtil.cloneQualifier(cell)));
            System.out.println("值:" + Bytes.toString(CellUtil.cloneValue(cell)));
        }
    }

    public static void main(String[] args) throws IOException {
       // System.out.println(isExitTable("student2"));
        //createTable("test1","info","message");
        //deleteTable("test1");
        addData("student","1003","info","age","20");
    }
}
